package com.ithuameng.cosy.mq.listener;

import com.ithuameng.cosy.constant.ChannelAttr;
import com.ithuameng.cosy.constants.Constants;
import com.ithuameng.cosy.model.Session;
import com.ithuameng.cosy.group.SessionGroup;
import com.ithuameng.cosy.model.Message;
import com.ithuameng.cosy.util.JSONUtils;
import com.ithuameng.cosy.util.RedisCache;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

@Slf4j
@Component
public class BindMessageListener {

    @Resource
    private RedisCache redisCache;

    /**
     * 一个账号只能在同一个类型的终端登录
     * 如: 多个android或ios不能同时在线
     * 一个android或ios可以和web，桌面同时在线
     */
    private final Map<String, String[]> conflictMap = new HashMap<>();

    @Resource
    private SessionGroup sessionGroup;

    public BindMessageListener() {
        conflictMap.put(Session.CHANNEL_ANDROID, new String[]{Session.CHANNEL_ANDROID, Session.CHANNEL_IOS});
        conflictMap.put(Session.CHANNEL_IOS, new String[]{Session.CHANNEL_ANDROID, Session.CHANNEL_IOS});
        conflictMap.put(Session.CHANNEL_WINDOWS, new String[]{Session.CHANNEL_WINDOWS, Session.CHANNEL_WEB, Session.CHANNEL_MAC});
        conflictMap.put(Session.CHANNEL_WEB, new String[]{Session.CHANNEL_WINDOWS, Session.CHANNEL_WEB, Session.CHANNEL_MAC});
        conflictMap.put(Session.CHANNEL_MAC, new String[]{Session.CHANNEL_WINDOWS, Session.CHANNEL_WEB, Session.CHANNEL_MAC});
    }

    @RabbitListener(queues = Constants.BIND_MESSAGE_INNER_QUEUE)
    public void bindMessageListener(com.rabbitmq.client.Channel rabbitChannel, org.springframework.amqp.core.Message message) throws IOException {

        // 判断此消息是否已经消费过
        String msgId = (String) message.getMessageProperties().getHeaders().get("spring_returned_message_correlation");

        long tag = message.getMessageProperties().getDeliveryTag();

        if (redisCache.getCacheMap(Constants.MESSAGE_CONSUME_REDIS_RECORD).containsKey(msgId)) {

            log.info("[" + msgId + "] message has been consumed from BindMessageListener.");

            rabbitChannel.basicAck(tag, false);

            return;
        }

        try {
            Session session = JSONUtils.fromJson(message.getBody(), Session.class);

            String uid = session.getUid();
            String[] conflictChannels = conflictMap.get(session.getChannel());

            // 获取当前uid登录的所有设备
            Collection<Channel> channelList = sessionGroup.find(uid, conflictChannels);

            // 将当前登录设备移除再外
            channelList.removeIf(channel -> session.getNid().equals(channel.attr(ChannelAttr.ID).get()));

            // 获取到其他在线的终端连接，提示账号再其他终端登录
            channelList.forEach(channel -> {

                // 同一个设备直接关闭channal
                if (Objects.equals(session.getDeviceId(), channel.attr(ChannelAttr.DEVICE_ID).get())) {
                    channel.close();
                    return;
                }

                // 其它设备退出登录 例如：chrome先登录，ie登录之后chrome则退出
                Message msg = new Message();
                msg.setAction(Constants.FORCE_OFFLINE_ACTION);
                msg.setReceiver(uid);
                msg.setSender(Constants.SYSTEM_ID);
                msg.setContent(session.getDeviceName());
                channel.writeAndFlush(msg);
                channel.close();
            });
        } catch (Exception e) {

            // 消息处理失败 true阻止消息重新回到队列中再次消费
            log.info("[" + msgId + "] message processing failure from BindMessageListener : " + e.getMessage());

            rabbitChannel.basicNack(tag, false, true);
        }

        // 标记消息被处理过
        Map<String, String> data = new HashMap<>(1);

        data.put(msgId, Constants.BIND_MESSAGE_ROUTING_KEY);

        redisCache.setCacheMap(Constants.MESSAGE_CONSUME_REDIS_RECORD, data);

        rabbitChannel.basicAck(tag, false);
    }
}
